#include "config.h"

#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>

#define DBG_SUBSYS S_LIBSTORAGE

#include "adt.h"
#include "ynet_rpc.h"
#include "core.h"
#include "md_map.h"
#include "lich_md.h"
#include "md_parent.h"
#include "cache.h"
#include "ylog.h"
#include "dbg.h"

typedef struct {
        chkid_t chkid;
        chkid_t parent;
} entry_t;

static mcache_t *cache;

static int __entry_new(entry_t **_ent, const chkid_t *chkid, const chkid_t *parent)
{
        int ret;
        entry_t *ent;

        ret = ymalloc((void**)&ent, sizeof(entry_t));
        if (unlikely(ret)) 
                GOTO(err_ret, ret);

        ent->chkid = *chkid;
        ent->parent = *parent;
        *_ent = ent;

        return 0;
err_ret:
        return ret;
}

static void __entry_free(entry_t *ent)
{
        yfree((void **)&ent);
}

static void __md_parent_release(mcache_entry_t *cent)
{
        mcache_release(cent);
}

static int __md_parent_get(const chkid_t *chkid, chkid_t *parent)
{
        int ret, retry = 0;
        nid_t nid;

retry:
        ret = md_map_getsrv(chkid, &nid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        DINFO("get parent of "CHKID_FORMAT"\n", CHKID_ARG(chkid));

        ret = md_getparent(&nid, chkid, parent);
        if (unlikely(ret)) {
                if ((ret == EREMCHG || ret == ENOENT) && retry == 0) {
                        md_map_drop(chkid, &nid);
                        retry = 1;
                        goto retry;
                } else
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int md_parent_get(const chkid_t *chkid, chkid_t *parent)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        if (chkid->type == __RAW_CHUNK__ || chkid->type == __VOLUME_SUB_CHUNK__ ) {
                cid2fid(parent, chkid);
                goto out;
        }

        if (chkid->type == __POOL_SUB_CHUNK__) {
                cid2fid(parent, chkid);
                parent->type = __POOL_CHUNK__;
                goto out;
        }

        YASSERT(chkid->type == __POOL_CHUNK__
                || chkid->type == __VOLUME_CHUNK__);

retry:
        ret = mcache_get(cache, chkid, &cent);
        if (unlikely(ret)) {
                if (ret == ENOENT) {
                        DBUG("chunk "CHKID_FORMAT" get\n", CHKID_ARG(chkid));

                        ret = __md_parent_get(chkid, parent);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);

                        ret = __entry_new(&ent, chkid, parent);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);

                        ret = mcache_insert(cache, chkid, ent);
                        if (unlikely(ret)) {
                                __entry_free(ent);
                                if (ret == EEXIST)
                                        goto retry;
                                else
                                        GOTO(err_ret, ret);
                        }

                        goto retry;
                } else
                        GOTO(err_ret, ret);
        }


        ret = mcache_wrlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;
        *parent = ent->parent;

        mcache_unlock(cent);

        __md_parent_release(cent);

out:
        YASSERT(!chkid_isnull(parent));
        
        return 0;
err_release:
        __md_parent_release(cent);
err_ret:
        return ret;
}

int md_parent_get1(const chkid_t *chkid, chkid_t *parent)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        if (chkid->type == __RAW_CHUNK__ || chkid->type == __VOLUME_SUB_CHUNK__ ) {
                cid2fid(parent, chkid);
                goto out;
        }

        if (chkid->type == __POOL_SUB_CHUNK__) {
                cid2fid(parent, chkid);
                parent->type = __POOL_CHUNK__;
                goto out;
        }

        YASSERT(chkid->type == __POOL_CHUNK__
                || chkid->type == __VOLUME_CHUNK__);

        ret = mcache_get(cache, chkid, &cent);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = mcache_rdlock(cent);
        if (unlikely(ret))
                GOTO(err_release, ret);

        ent = cent->value;
        *parent = ent->parent;

        mcache_unlock(cent);

        __md_parent_release(cent);

out:
        YASSERT(!chkid_isnull(parent));

        return 0;
err_release:
        __md_parent_release(cent);
err_ret:
        return ret;
}

int md_parent_update(const chkid_t *chkid, const chkid_t *parent)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;

        YASSERT(!chkid_isnull(parent));
        YASSERT((parent->type == __POOL_CHUNK__
                 || chkid->type == __VOLUME_CHUNK__)
                && parent->id != 0);
        YASSERT(chkid->type == __POOL_CHUNK__
                || chkid->type == __VOLUME_CHUNK__);

retry:
        ret = mcache_get(cache, chkid, &cent);
        if (unlikely(ret)) {
                if (ret == ENOENT) {
                        DBUG("chunk "CHKID_FORMAT" update\n", CHKID_ARG(chkid));
                        ret = __entry_new(&ent, chkid, parent);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);

                        ret = mcache_insert(cache, chkid, ent);
                        if (unlikely(ret)) {
                                __entry_free(ent);
                                if (ret == EEXIST)
                                        goto retry;
                                else
                                        GOTO(err_ret, ret);
                        }
                } else
                        GOTO(err_ret, ret);
        } else {
                ret = mcache_wrlock(cent);
                if (unlikely(ret))
                        GOTO(err_release, ret);

                ent = cent->value;
                ent->parent = *parent;

                mcache_unlock(cent);
                __md_parent_release(cent);
        }

        return 0;
err_release:
        __md_parent_release(cent);
err_ret:
        return ret;
}

int md_parent_drop(const chkid_t *chkid, const chkid_t *parent)
{
        int ret;
        mcache_entry_t *cent;
        entry_t *ent;
        chkid_t newid;

        YASSERT(chkid->type == __POOL_CHUNK__
                || chkid->type == __VOLUME_CHUNK__);

        ret = mcache_get(cache, chkid, &cent);
        if (unlikely(ret)) {
                if (ret == ENOENT) {
                        //XXX
                } else
                        GOTO(err_ret, ret);
        } else {
                ent = cent->value;

                if (chkid_cmp(&ent->parent, parent) == 0) {
                        ret = __md_parent_get(chkid, &newid);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);

                        ret = mcache_wrlock(cent);
                        if (unlikely(ret))
                                GOTO(err_release, ret);

                        ent->parent = newid;

                        mcache_unlock(cent);
                }

                __md_parent_release(cent);
        }

        return 0;
err_release:
        __md_parent_release(cent);
err_ret:
        return ret;
}

static int __cmp(const void *s1, const void *s2)
{
        const chkid_t *id1 = s1;
        const chkid_t *id2 = &((entry_t *)s2)->chkid;

        DBUG("cmp "CHKID_FORMAT" : "CHKID_FORMAT"\n",
             CHKID_ARG(id1), CHKID_ARG(id2));

        return !chkid_cmp(id1, id2);
}

static uint32_t __hash(const void *key)
{
        const chkid_t *id = key;

        return id->id;
}

static uint32_t __core_hash(const void *key)
{
        const chkid_t *id = key;

        return core_hash(id);
}

static int __drop(void *value, mcache_entry_t *cent, int recycle)
{
        entry_t *ent;

        (void) cent;
        ent = (entry_t *)value;

        if (ent) {
                if (recycle) {
                        entry_t *entry = value;
                        DINFO("recycle "CHKID_FORMAT"\n", &entry->chkid);
                }
                yfree((void **)&ent);
        }

        return 0;
}

int md_parent_init()
{
        int ret;

        ret = mcache_init(&cache, MCACHE_SIZE, __cmp, __hash, __core_hash,
                          __drop, 0, "md_parent");
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}
